package com.czl.dxpro.dx_02_match;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.TreeMap;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.czl.dxpro.dx_04_utils.StringComparator;
import com.czl.dxpro.dx_04_utils.TProperties;

public class DxMatchMapper extends
		Mapper<LongWritable, Text, NullWritable, Text> {

	private Map<String, TreeMap<String, String>> joinData = new HashMap<String, TreeMap<String, String>>();

	// setup方法用来预加载附加的文件的————t_dx_basic_msg_addr.txt
	@Override
	protected void setup(
			Mapper<LongWritable, Text, NullWritable, Text>.Context context)
			throws IOException, InterruptedException {
		// reader读入t_dx_basic_msg_addr.txt
		Path file = new Path(context.getCacheFiles()[0].getPath());
		BufferedReader reader = new BufferedReader(new FileReader(
				file.toString()));
		String str = null;
		try {
			while ((str = reader.readLine()) != null) {
				// 分割缓存中的表
				String[] splits = str.split(TProperties
						.getValue("fileoutsplit"));
				// 当已经添加过该一级域名时,则直接获取该域名的treeMap集合添加数据即可
				if (joinData.containsKey(splits[1])) {
					// 增加treemap中url值：匹配地址,匹配级别,行为ID,产品标识,预购类型
					joinData.get(splits[1]).put(
							splits[2] + "," + splits[3] + "," + splits[0] + ","
									+ splits[4] + "," + splits[5], "");
				} else {
					TreeMap<String, String> treemap = new TreeMap<String, String>(
							new StringComparator());
					// url地址,匹配级别,行为ID,是否产品,预购类型
					treemap.put(splits[2] + "," + splits[3] + "," + splits[0]
							+ "," + splits[4] + "," + splits[5], "");
					joinData.put(splits[1], treemap);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			reader.close();
		}

	}

	// 负责读入数据源数据进行分析处理的————dxclear处理结果文档的DxClear/part-m-000*
	@Override
	protected void map(LongWritable key, Text value,
			Mapper<LongWritable, Text, NullWritable, Text>.Context context)
			throws IOException, InterruptedException {
			//获取value中的一级域名，
			String[] values = value.toString().split(TProperties.getValue("fileoutsplit"));
			//value[1]就是一级域名
			//然后根据此域名从joinData中取数据，
			if (joinData.containsKey(values[1])) {
				// 获取一级域名下匹配地址
				TreeMap<String, String> treeMap = joinData.get(values[1]);
				Iterator iterator = treeMap.keySet().iterator();
				// 按级别循环匹配
				while (iterator.hasNext()) {
					String[] krule = ((String) iterator.next()).split(",");
					if (values[2].indexOf(krule[0]) >= 0) {
						//并配合value其他分割数据，重新组合新的输出结果
						StringBuffer sb = new StringBuffer();
						// 行为id，用户号码，是否产品，url,预购类型
						sb.append(krule[2]).append(TProperties.getValue("outfilesplit")).append(values[0])
								.append(TProperties.getValue("outfilesplit")).append(krule[3])
								.append(TProperties.getValue("outfilesplit")).append(values[2])
								.append(TProperties.getValue("outfilesplit")).append(krule[4]);
						Text ovalue = new Text(sb.toString());
						// 匹配后输出并退出匹配循环
						context.write(NullWritable.get(), ovalue);
						return;
					}
				}
			}

			
	}

}
